package com.yujiahao.bigdata.rdd.insterance

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
    TODO 将文件作为数据源，构建数据模型RDD--研究分区
      1、textFile和whlieTestFile方法参数说明：
        >第一个参数表示文件（目录）路径。
        >第二个参数表示最小分区数量，参数存在默认值是可以不需要传递的。
          =>默认值：def defaultMinPartitions: Int = math.min(defaultParallelism, 2);
          =>源码参数解读:参数一：defaultParallelism表示 如果有配置参数就用配置参数的分区，如果配置对象中没有配置相关参数，那么会采用当前环境的最大虚拟核数。
          =>参数二：表示2个分区--详情见下面说明
      2、关于分区的说明
      Spark的文件操作是没有的，底层采用的是Hadoop的文件操作，所以要弄清楚有多少分区，要明白Hadoop中文件如何切片的
      3、源码解析：
          【totalSize】：文件的总大小（字节数）
          =>源码：totalSize += file.getLen();
          【goalSize】  : 每个分区预计的字节大小
          =>源码：long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
          【分区数量】： totalSize / goalSize
      【特别的】：如果分区数量有余数，那么这个余数如果大于每个分区预计字节数的10%，那么最终结果要加一
         =>源码：private val SPLIT_SLOP: Double = 1.1// 10% slop
        4、Hadoop计算分区和读取分区数据的方式不一样。
           1. 计算分区是按照字节来计算的
           2. 读取分区数据不是按照字节，是按照行来读取的（单词不能跨行）
              读取数据时，按照偏移量来计算，相同的偏移量不能被重复读取
        5、 RDD的弹性可以体现在分区的变化:后面详细讲
         默认情况下，分区数据是不均衡的，可以在任务的执行的过程中，改变分区，让数据更加均衡。
         改变分区等同于改变数据的位置，这个需要和分区器发生关联
        6、大数据中一行就是一个数据
 */

object Spark01_RDD_Instance_Disk_Partition {
  def main(args: Array[String]): Unit = {
    //TODO 1、构建环境--获得连接
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 将文件作为数据源，构建数据模型RDD--研究分区
    val rdd: RDD[String] = sc.textFile("data/word.txt", 3)
/*案例：分区的过程分析：
        todo 1. 计算分区是按照字节来计算的
          yujiahao@@
          yu@@
          ha1@@
          totalsize   => 17
          goal        => 17 / 3=5..
          partition   => 17 /5 =3+1=4
        todo 2. 读取分区数据不是按照字节，是按照行来读取的（单词不能跨行）
            读取数据时，按照偏移量来计算，相同的偏移量不能被重复读取
          [0,5]   =>yujiahao
          [5,9]   =>yu
          [9,13]  =>ha1
          [13,17]  =>
          */
    rdd.saveAsTextFile("output")
    rdd.collect().foreach(println)
    //TODO 5、释放资源--关闭连接
    sc.stop()

  }

}
